package com.ruyuan.engine.service;
import com.ruyuan.engine.cache.CacheManager;
import com.ruyuan.engine.cache.impl.RedisCacheManagerImpl;
import com.ruyuan.engine.pojo.CatchData;
import com.ruyuan.engine.pojo.CombinationRuleParam;
import com.ruyuan.engine.pojo.RuleEventParam;
import com.ruyuan.engine.utils.Configuration;
import com.ruyuan.engine.utils.RuleCommonProcessUtils;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * 第三版clickhouseServer查询服务：
 *
 * */
public class ClickHouseServer {
    Connection connection;
    CacheManager cacheManager;
    Long REDIS_TTL;
    public ClickHouseServer(Connection connection) {
        this.connection = connection;
        cacheManager = new RedisCacheManagerImpl();
        Config load = ConfigFactory.load();
        REDIS_TTL = load.getLong(Configuration.REDIS_CACHE_TTL);
    }

    /**
     * 返回满足条件的规则结果字符串
     * */
    private String mathRuleCombinationConditionCountToStr(String userId, CombinationRuleParam combinationRuleParam,Long startTime,Long endTime) throws SQLException {
        String sql = combinationRuleParam.getSql();
        PreparedStatement preparedStatement = connection.prepareStatement(sql);
        preparedStatement.setString(1,userId);
        preparedStatement.setLong(2,startTime);
        preparedStatement.setLong(3,endTime);

        List<RuleEventParam> ruleEventParams = combinationRuleParam.getRuleEventParams();

        /**
         * 获取所有的事件类型
         * List<String> A1 C2 D3 F5 事件
         * */

        List<String> eventIds = ruleEventParams.stream().map(e -> e.getEventId()).collect(Collectors.toList());

        //和clickhouse查询的结果进行匹配
        ResultSet resultSet = preparedStatement.executeQuery();
        StringBuilder stringBuilder = new StringBuilder();
        while (resultSet.next()) {
            String eventId = resultSet.getString(1);
            //eventId在eventIds里面是否存在，存在的话就返回下标
            stringBuilder.append((eventIds.indexOf(eventId)+1));
        }

        //A1 C2 D3 A1 F5 D3-> 103103
        return stringBuilder.toString();
    }

    /**
     * 查询该组合规则出现的次数
     * */
    public Integer mathRuleCombinationConditionCount(String userId,CombinationRuleParam combinationRuleParam,Long startTime,Long endTime) throws SQLException {
        /**
         * 先查询redis缓存中的结果
         * 缓存的结果在什么情况下是有效的：
         *      缓存存储的数据的时间范围：[A3 -> A6]
         *      查询条件的时间范围：
         *          [A3 -> A6] 直接用缓存的结果返回就可以了
         *          [A3 -> A12] 判断缓存数据的结果值 是否 >= 规则条件的设定值，如果满足返回缓存结果，不满足 使用缓存的结果 [A3 -> A6] + [A6 ->A12] 作为整个结果返回
         *          [A1 -> A6]  判断缓存数据的结果值 是否 >= 规则条件的设定值，如果满足返回缓存结果，不满足 [A1 - A3]+缓存结果
         *          [A1 -> A10] 判断缓存数据的结果值 是否 >= 规则条件的设定值，如果满足返回缓存结果，不满足 缓存的结果就无效了
         *
         * */
        String primaryKey = RuleCommonProcessUtils.generateKey(userId, combinationRuleParam.getCacheId());
        Long currentTime = System.currentTimeMillis();
        CatchData cacheData = cacheManager.getCacheData(RuleCommonProcessUtils.generateKey(userId, combinationRuleParam.getCacheId()));
        //遍历缓存的结果集
        Set<String> keys = cacheData.getRedisCacheMap().keySet();
        for(String key:keys) {
            String[] mapKeys = key.split(":");

            /**
             * 针对过期的key，执行删除的操作，判断key过期的依据：保存到redis的时间距离当前时间是否超过6个小时，
             *     如果超过就删除
             * */
            //获取缓存插入的时间
            long catchInsertTime = Long.parseLong(mapKeys[2]);
            if (currentTime - catchInsertTime >= REDIS_TTL) {
                //删除缓存的数据
                cacheManager.delCacheDataByMapKey(primaryKey,key);
            }


            long catchStartTime = Long.parseLong(mapKeys[0]);
            long catchEndTime = Long.parseLong(mapKeys[1]);
            String catchResult = cacheData.getRedisCacheMap().get(key);

            //[A3 -> A6](查询范围相同) 直接用缓存的结果返回就可以了
            if (catchStartTime == startTime && catchEndTime ==endTime) {
                /**
                 * 从缓存中查询到了数据，需要做update操作
                 * 1. 先删除之前的缓存数据
                 * 2. 更新缓存数据
                 *
                 * A3  ->   A6
                 * 匹配缓存的数据：A3 -> A6
                 * */
                updateCatch(primaryKey, currentTime, key, catchStartTime, catchEndTime, catchResult);
                return RuleCommonProcessUtils.sequenceMathCountByRegex(catchResult,combinationRuleParam.getRegexp());
            }

            //[A3 -> A12] => 使用缓存的结果 [A3 -> A6] + [A6 ->A12] 作为整个结果返回
            if (startTime == catchStartTime && endTime >= catchEndTime) {
                int catChCount = RuleCommonProcessUtils.sequenceMathCountByRegex(catchResult, combinationRuleParam.getRegexp());
                int minCount = combinationRuleParam.getMinCount();
                if (catChCount >= minCount) {
                    return catChCount;
                }else {
                    //查询clickhouse
                    String clickHouseResult = mathRuleCombinationConditionCountToStr(userId, combinationRuleParam, catchEndTime, endTime);
                    /**
                     *  添加缓存的操作
                     * 查询时间区间： A3  ->   A6
                     * 需要匹配时间区间的数据：A3 -> A12
                     *
                     * 1. 清空之前的缓存数据
                     * 2. 把三段需要缓存的数据添加到HashMap
                     * 3. 存储HashMap到redis
                     * */
                    cacheManager.delCacheDataByMapKey(primaryKey,key);
                    HashMap concurrentCacheMap = new HashMap();
                    //更新原有缓存的数据
                    concurrentCacheMap.put(RuleCommonProcessUtils.generateKey(catchStartTime,catchEndTime,currentTime),catchResult);
                    //对从clickhouse里面查询的数据进行缓存,缓存第二个部分的数据
                    concurrentCacheMap.put(RuleCommonProcessUtils.generateKey(catchEndTime,endTime,currentTime),clickHouseResult);
                    //对clickhouse和缓存中的数据进行保存
                    concurrentCacheMap.put(RuleCommonProcessUtils.generateKey(catchStartTime,endTime,currentTime),catchResult+clickHouseResult);
                    cacheManager.addCacheData(primaryKey,concurrentCacheMap);
                    return RuleCommonProcessUtils.sequenceMathCountByRegex(catchResult + clickHouseResult, combinationRuleParam.getRegexp());
                }
            }

            //[A1 -> A6]
            if (endTime == catchEndTime && startTime < catchStartTime) {
                int catChCount = RuleCommonProcessUtils.sequenceMathCountByRegex(catchResult, combinationRuleParam.getRegexp());
                int minCount = combinationRuleParam.getMinCount();
                if (catChCount >= minCount) {
                    return catChCount;
                }else {
                    //查询clickhouse
                    String clickHouseResult = mathRuleCombinationConditionCountToStr(userId, combinationRuleParam, startTime, catchStartTime);
                    /**
                     * 将查询结果添加到缓存
                     * */
                    cacheManager.delCacheDataByMapKey(primaryKey,key);
                    HashMap concurrentCacheMap = new HashMap();
                    //更新原有缓存的数据
                    concurrentCacheMap.put(RuleCommonProcessUtils.generateKey(catchStartTime,catchEndTime,currentTime),catchResult);
                    //对从clickhouse里面查询的数据进行缓存,缓存第二个部分的数据
                    concurrentCacheMap.put(RuleCommonProcessUtils.generateKey(startTime,catchStartTime,currentTime),clickHouseResult);
                    //对clickhouse和缓存中的数据进行保存
                    concurrentCacheMap.put(RuleCommonProcessUtils.generateKey(startTime,endTime,currentTime),catchResult+clickHouseResult);
                    cacheManager.addCacheData(primaryKey,concurrentCacheMap);
                    return RuleCommonProcessUtils.sequenceMathCountByRegex(catchResult + clickHouseResult, combinationRuleParam.getRegexp());
                }
            }

            //包含了缓存的区间
            if (endTime >= catchEndTime && startTime <= catchEndTime) {
                int catChCount = RuleCommonProcessUtils.sequenceMathCountByRegex(catchResult, combinationRuleParam.getRegexp());
                int minCount = combinationRuleParam.getMinCount();
                if (catChCount >= minCount) {
                    updateCatch(primaryKey, currentTime, key, catchStartTime, catchEndTime, catchResult);
                    return catChCount;
                }
            }
        }
        String resultSequence = mathRuleCombinationConditionCountToStr(userId, combinationRuleParam, startTime, endTime);
        //然后通过正则匹配结果
        /**
         * 将结果数据放入redis缓存当中
         * resultSequence
         * */
        HashMap<String,String> concurrentCacheMap = new HashMap<>();
        concurrentCacheMap.put(RuleCommonProcessUtils.generateKey(startTime,endTime,currentTime),resultSequence);
        cacheManager.addCacheData(primaryKey,concurrentCacheMap);
        return RuleCommonProcessUtils.sequenceMathCountByRegex(resultSequence, combinationRuleParam.getRegexp());
    }

    private void updateCatch(String primaryKey, Long currentTime, String key, long catchStartTime, long catchEndTime, String catchResult) {
        cacheManager.delCacheDataByMapKey(primaryKey, key);
        //生成新的缓存数据更新
        HashMap<String, String> concurrentCacheMap = new HashMap<>();
        concurrentCacheMap.put(catchStartTime + ":" + catchEndTime + ":" + currentTime, catchResult);
        cacheManager.addCacheData(primaryKey, concurrentCacheMap);
    }

    /**
     * 查询组合规则并且返回匹配结果和次数。
     * Tuple2<String,Integer>
     * */
    public Tuple2<String,Integer> mathRuleCombinationConditionCount(String userId, CombinationRuleParam combinationRuleParam, Long startTime, Long endTime,Boolean flag) throws SQLException {
        String primaryKey = RuleCommonProcessUtils.generateKey(userId, combinationRuleParam.getCacheId());
        Long currentTime = System.currentTimeMillis();
        CatchData cacheData = cacheManager.getCacheData(RuleCommonProcessUtils.generateKey(userId, combinationRuleParam.getCacheId()));
        Set<String> keys = cacheData.getRedisCacheMap().keySet();
        for(String key:keys) {
            String[] mapKeys = key.split(":");
            long catchStartTime = Long.parseLong(mapKeys[0]);
            long catchEndTime = Long.parseLong(mapKeys[1]);
            long catchInsertTime = Long.parseLong(mapKeys[2]);
            if (currentTime - catchInsertTime >= REDIS_TTL) {
                cacheManager.delCacheDataByMapKey(primaryKey,key);
            }

            String catchResult = cacheData.getRedisCacheMap().get(key);
            int countCache = RuleCommonProcessUtils.sequenceMathCountByRegex(catchResult, combinationRuleParam.getRegexp());

            if (catchStartTime == startTime && catchEndTime ==endTime) {
                updateCatch(primaryKey, currentTime, key, catchStartTime, catchEndTime, catchResult);
                return Tuple2.of(catchResult,countCache);
            }

            if (startTime == catchStartTime && endTime >= catchEndTime) {
                int minCount = combinationRuleParam.getMinCount();
                /**
                 * 这里可能返回的不是完整的结果，而是结果的一部分缓存
                 * flag：是否需要返回完整的结果。
                 *  后面需要使用 完成的结果+flinkState里面数据做匹配得到最终的结果。
                 * */
                if (countCache >= minCount && !flag) {
                    return Tuple2.of(catchResult,countCache);
                }else {
                    //查询clickhouse
                    String clickHouseResult = mathRuleCombinationConditionCountToStr(userId, combinationRuleParam, catchEndTime, endTime);
                    cacheManager.delCacheDataByMapKey(primaryKey,key);
                    if (clickHouseResult != null && clickHouseResult.length() != 0) {
                        HashMap concurrentCacheMap = new HashMap();
                        //更新原有缓存的数据
                        concurrentCacheMap.put(RuleCommonProcessUtils.generateKey(catchStartTime,catchEndTime,currentTime),catchResult);
                        //对从clickhouse里面查询的数据进行缓存,缓存第二个部分的数据
                        concurrentCacheMap.put(RuleCommonProcessUtils.generateKey(catchEndTime,endTime,currentTime),clickHouseResult);
                        //对clickhouse和缓存中的数据进行保存
                        concurrentCacheMap.put(RuleCommonProcessUtils.generateKey(catchStartTime,endTime,currentTime),catchResult+clickHouseResult);
                        cacheManager.addCacheData(primaryKey,concurrentCacheMap);
                    }
                    int count = RuleCommonProcessUtils.sequenceMathCountByRegex(catchResult + clickHouseResult, combinationRuleParam.getRegexp());
                    return Tuple2.of(catchResult+clickHouseResult,count);
                }
            }

            //[A1 -> A6]
            if (endTime == catchEndTime && startTime < catchStartTime) {
                int minCount = combinationRuleParam.getMinCount();
                //这里可能返回的不是完整的结果，而是结果的一部分缓存
                if (countCache >= minCount && !flag) {
                    return Tuple2.of(catchResult,countCache);
                }else {
                    //查询clickhouse
                    String clickHouseResult = mathRuleCombinationConditionCountToStr(userId, combinationRuleParam, startTime, catchStartTime);
                    cacheManager.delCacheDataByMapKey(primaryKey,key);
                    if (clickHouseResult != null && clickHouseResult.length() != 0) {
                        HashMap concurrentCacheMap = new HashMap();
                        //更新原有缓存的数据
                        concurrentCacheMap.put(RuleCommonProcessUtils.generateKey(catchStartTime,catchEndTime,currentTime),catchResult);
                        //对从clickhouse里面查询的数据进行缓存,缓存第二个部分的数据
                        concurrentCacheMap.put(RuleCommonProcessUtils.generateKey(startTime,catchStartTime,currentTime),clickHouseResult);
                        //对clickhouse和缓存中的数据进行保存
                        concurrentCacheMap.put(RuleCommonProcessUtils.generateKey(startTime,endTime,currentTime),catchResult+clickHouseResult);
                        cacheManager.addCacheData(primaryKey,concurrentCacheMap);
                    }
                    int count = RuleCommonProcessUtils.sequenceMathCountByRegex(catchResult + clickHouseResult, combinationRuleParam.getRegexp());
                    return Tuple2.of(catchResult+clickHouseResult,count);
                }
            }

            //包含了缓存的区间
            if (endTime >= catchEndTime && startTime <= catchEndTime) {
                int minCount = combinationRuleParam.getMinCount();
                if (countCache >= minCount && !flag) {
                    updateCatch(primaryKey, currentTime, key, catchStartTime, catchEndTime, catchResult);
                    return Tuple2.of(catchResult,countCache);
                }
            }
        }
        String resultSequence = mathRuleCombinationConditionCountToStr(userId, combinationRuleParam, startTime, endTime);
        //然后通过正则匹配结果
        /**
         * 将结果数据放入redis缓存当中
         * resultSequence
         * */
        if (resultSequence != null && resultSequence.length() !=0) {
            HashMap<String,String> concurrentCacheMap = new HashMap<>();
            concurrentCacheMap.put(RuleCommonProcessUtils.generateKey(startTime,endTime,currentTime),resultSequence);
            cacheManager.addCacheData(primaryKey,concurrentCacheMap);
        }
        int count = RuleCommonProcessUtils.sequenceMathCountByRegex(resultSequence, combinationRuleParam.getRegexp());
        return Tuple2.of(resultSequence,count);
    }

    public Tuple2<String,Integer> mathRuleCombinationConditionCountByFalse(String userId, CombinationRuleParam combinationRuleParam, Long startTime, Long endTime) throws SQLException {
        return mathRuleCombinationConditionCount(userId,combinationRuleParam,startTime,endTime,false);
    }
}
